package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.PaperQueBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author zhangsan
 * @Date 2022/10/20 10:26
 * @Description //TODO 考试域 试卷问题 相关情况 窗口
 */
public class DwsTestPaperQueWindow {
    public static void main(String[] args) throws Exception{
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点设置
        //TODO 3.从kafka的  dwd_test_paper_que_exam 主题中读取数据
        String topicId = "dwd_test_paper_que_exam";
        String groupId = "dws_test_paper_que_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topicId, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //TODO 4. 将 数据类型转换 jsonStr->jsonObj 再封装为实体类
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        SingleOutputStreamOperator<PaperQueBean> queBeanDS = jsonObjDS.map(
                new MapFunction<JSONObject, PaperQueBean>() {
                    @Override
                    public PaperQueBean map(JSONObject jsonObj) throws Exception {
                        String questionId = jsonObj.getString("question_id");
                        String userId = jsonObj.getString("user_id");
                        String isCorrect = jsonObj.getString("is_correct");
                        Long ts = jsonObj.getLong("ts");
                        PaperQueBean queBean = PaperQueBean.builder()
                                .questionId(questionId)
                                .userId(userId)
                                .isCorrect(Long.valueOf(isCorrect))
                                .queCnt(1L)
                                .queCorrectCnt(Long.valueOf(isCorrect))
                                .userIdSet(
                                        new HashSet<String>(
                                                Collections.singleton(userId)
                                        )
                                )
                                .correctUserSet(
                                        "1".equals(isCorrect) ?
                                                new HashSet<String>(Collections.singleton(userId))
                                                : new HashSet<String>()
                                )
                                .ts(ts)
                                .build();
                        return queBean;
                    }
                }
        );

        //TODO 5.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<PaperQueBean> withWatermarkDS = queBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<PaperQueBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<PaperQueBean>() {
                                    @Override
                                    public long extractTimestamp(PaperQueBean paperQueBean, long l) {
                                        return paperQueBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 6.按照 问题id进行分组 question_id
        KeyedStream<PaperQueBean, String> queIdKeyedDS = withWatermarkDS.keyBy(PaperQueBean::getQuestionId);
        //TODO 7.开窗
        WindowedStream<PaperQueBean, String, TimeWindow> windowDS = queIdKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 8.聚合
        SingleOutputStreamOperator<PaperQueBean> reduceDS = windowDS.reduce(
                new ReduceFunction<PaperQueBean>() {
                    @Override
                    public PaperQueBean reduce(PaperQueBean value1, PaperQueBean value2) throws Exception {
                        value1.setQueCnt(value1.getQueCnt() + value2.getQueCnt());
                        value1.setQueCorrectCnt(value1.getQueCorrectCnt() + value2.getQueCorrectCnt());
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        value1.getCorrectUserSet().addAll(value2.getCorrectUserSet());
                        return value1;
                    }
                },
                new WindowFunction<PaperQueBean, PaperQueBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<PaperQueBean> input, Collector<PaperQueBean> out) throws Exception {
                        for (PaperQueBean queBean : input) {
                            queBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            queBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            queBean.setTs(System.currentTimeMillis());
                            queBean.setUniqueUserCnt((long) queBean.getUserIdSet().size());
                            queBean.setUniqueCorrectUserCnt((long) queBean.getCorrectUserSet().size());
                            out.collect(queBean);
                        }
                    }
                }
        );
        //TODO 9.关联一下维度表 问题维度表  DIM_TEST_QUESTION_INFO
        SingleOutputStreamOperator<PaperQueBean> withQueDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<PaperQueBean>("dim_test_question_info") {
                    @Override
                    public void join(PaperQueBean queBean, JSONObject dimInfoJsonObj) {
                        queBean.setQuestionName(dimInfoJsonObj.getString("question_text"));
                    }

                    @Override
                    public String getKey(PaperQueBean obj) {
                        return obj.getQuestionId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        //TODO 10.将聚合结果写到ck中
        withQueDS.print(">>>>>");
        withQueDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_test_paper_que_window values(?,?,?,?,?,?,?,?,?)"));
        env.execute();

    }
}
